package com.ksyun.campus.dataserver.services;

import com.alibaba.fastjson.JSONObject;
import com.ksyun.campus.dataserver.config.dataserverconfig.DataServerConfigProperties;
import com.ksyun.campus.dataserver.config.zookperconfig.ZookeeperConfigProperties;
import com.ksyun.campus.dataserver.domain.DataServerInfo;
import com.ksyun.campus.dataserver.util.MyZkSerializer;
import lombok.extern.slf4j.Slf4j;
import org.I0Itec.zkclient.ZkClient;
import org.apache.logging.log4j.util.Strings;
import org.apache.zookeeper.CreateMode;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Component
public class RegistService implements ApplicationRunner {
    @Resource
    private ZkClient zkClient;

    @Resource
    ZookeeperConfigProperties zookeeperConfigProperties;

    @Resource
    DataServerConfigProperties dataServerConfigProperties;

    @Resource
    private ServletWebServerApplicationContext webServerApplicationContext;

    @Resource
    private ConcurrentHashMap<String, DataServerInfo> dataServerMap;
    private String nodeName;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        registToCenter();
    }

    public void registToCenter() {
        // todo 将本实例信息注册至zk中心，包含信息 ip、port、capacity、rack、zone
        // 创建节点 记录信息
        InetAddress localHost;
        try {
            localHost = InetAddress.getLocalHost();
        } catch (UnknownHostException e) {
            throw new RuntimeException(e);
        }
        DataServerInfo dataServerInfo = new DataServerInfo(localHost.getHostAddress(), String.valueOf(webServerApplicationContext.getWebServer().getPort()), dataServerConfigProperties.getZone(), dataServerConfigProperties.getRack(),dataServerConfigProperties.getCapacity(), 0);

        String rootName = dataServerConfigProperties.getRootName();
        if (!Strings.isNotEmpty(rootName)) {
            throw new RuntimeException("根节点不能为空");
        }
        if (!zkClient.exists(rootName)) {
            zkClient.create(rootName, "root", CreateMode.PERSISTENT);
        }
        nodeName = rootName + dataServerConfigProperties.getNodeName();
        String dataServerInfoStr = JSONObject.toJSONString(dataServerInfo);
        // 解决zk写入数据乱码
        zkClient.setZkSerializer(new MyZkSerializer());
        nodeName = zkClient.createEphemeralSequential(nodeName, dataServerInfoStr);
        dataServerConfigProperties.setNodeName(nodeName.replace(dataServerConfigProperties.getRootName(), "")); // 设为创建得到的临时有序节点名
        // 将自己添加到本地节点列表中
        log.debug("node 已添加: {}", nodeName);
        log.debug("dataServerInfo: {}", dataServerInfoStr);
    }

    public Map<String, DataServerInfo> getDsList() {
        ConcurrentHashMap<String, DataServerInfo> dataServerMap = new ConcurrentHashMap<>();
        List<String> children = zkClient.getChildren(dataServerConfigProperties.getRootName());
        for (String node: children) {
            String dataServerInfoJsonStr = (String) zkClient.readData(dataServerConfigProperties.getRootName()+"/"+node);
            DataServerInfo server = JSONObject.parseObject(dataServerInfoJsonStr,DataServerInfo.class);
            dataServerMap.put(node, server);
        }
        return dataServerMap;
    }

    /**
     * 关闭前从zk中移除当前节点
     */
    @PreDestroy
    public void removeFromZkWhenShutdown() {
        zkClient.delete(nodeName);
        zkClient.close();
        log.debug("移除当前节点: {}", nodeName);
    }

}
